#ifndef __M_MSG_H__
#define __M_MSG_H__

#include "../mqcommon/logger.hpp"
#include "../mqcommon/helpper.hpp"
#include "../mqcommon/msg.pb.h"

#include <iostream>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <cassert>
#include <list>

namespace mmq
{
    #define DATAFILE_SUBFIX ".mqd"
    #define TMPFILE_SUBFIX ".mqd.tmp"

    using MessagePtr = std::shared_ptr<mmq::Message>;

    class MessageMapper
    {
    public:
        MessageMapper(std::string &basedir, const std::string &qname) : _qname(qname)
        {
            if(basedir.back() != '/') basedir += "/";
            if(FileHelper(basedir).Exists() == false)
            {
                assert(FileHelper::CreateDirectory(basedir));
            }
            _datafile = basedir + _qname + DATAFILE_SUBFIX;
            _tmpfile = basedir + _qname + TMPFILE_SUBFIX;
            CreateMsgFile();
        }

        bool CreateMsgFile()
        {
            if(FileHelper(_datafile).Exists()) return true;

            bool ret = FileHelper::CreateFile(_datafile);
            if(ret == false)
            {
                DLOG("创建队列数据文件 %s 失败! ", _datafile.c_str());
                return false;
            }
            return true;
        }

        void RemoveMsgFile()
        {
            FileHelper::RemoveFile(_datafile);
            FileHelper::RemoveFile(_tmpfile);
        }

        // 把数据写入到文件中
        bool Insert(MessagePtr &msg)
        {   
            Insert(_datafile, msg);
        }

        // 删除一条消息，并不是真正从文件中删除，而是把文件中该消息的有效标志置为"0"
        bool Remove(MessagePtr &msg)
        {
            // 1. 将msg中的有效标志位修改为"0"
            // payload()无法修改(因为返回的是const成员对象)，使用mutable_payload()(返回一个指针)
            msg->mutable_payload()->set_valid("0");
            // 2. 对msg进行序列化
            std::string body = msg->payload().SerializeAsString();
            if(body.size() != msg->length())
            {
                DLOG("不能修改文件中的数据信息，因为新生成的数据与原数据长度不一致! ");
                return false;
            }
            // 3. 将序列化后的数据，写入文件中的指定位置(覆盖原有数据)
            FileHelper _df_helper(_datafile);
            bool ret = _df_helper.Write(&body[0], msg->offset(), body.size());
            if(ret == false)
            {
                DLOG("向队列数据文件写入数据失败！");
                return false;
            }
            // DLOG("确认消息后，删除持久化消息成功：%s", msg->payload().body().c_str());
            return true;
        }

        // 垃圾数据回收，当数据量超过2000，并且超过一半都是无效数据时，进行垃圾回收。返回有效数据
        std::list<MessagePtr> GC()
        {
            // 1. 先从_datafile中读取出有效数据
            std::list<MessagePtr> result;
            bool ret = Load(result);
            if(ret == false)
            {
                DLOG("加载有效数据失败！");
                return result;
            }
            // DLOG("垃圾回收得到有效消息数量：%ld", result.size());
            // 2. 将有效数据写入_tmpfile
            FileHelper::CreateFile(_tmpfile);
            for(auto &msg : result)
            {
                DLOG("向临时文件写入数据：%s", msg->payload().body().c_str());
                ret = Insert(_tmpfile, msg);
                if(ret == false)
                {
                    DLOG("向临时文件写入消息数据失败！");
                    return result;
                }
            }
            // DLOG("垃圾回收后，向临时文件写入数据完毕，临时文件大小：%ld", FileHelper(_tmpfile).Size());
            // 3. 删除_datafile源文件
            ret = FileHelper::RemoveFile(_datafile);
            if(ret == false)
            {
                DLOG("删除源文件失败！");
                return result;   
            }
            // 4. 替换_datafile和_tmpfile的文件名
            ret = FileHelper(_tmpfile).Rename(_datafile);
            if(ret == false)
            {
                DLOG("修改临时文件名称失败！");
                return result;
            }
            return result;
        }
    private:
        bool Load(std::list<MessagePtr> &result)
        {
            // 加载出文件中所有有效数据，存储格式：4字节长度|数据|4字节长度|数据
            FileHelper _df_helper(_datafile);
            size_t fsize = _df_helper.Size();
            size_t offset = 0;
            size_t msg_size;
            bool ret;
            // DLOG("准备加载持久化数据，当前文件大小：%ld", fsize);
            while(offset < fsize)
            {
                // 先读取size_t字节长度
                ret = _df_helper.Read((char*)&msg_size, offset, sizeof(size_t));
                if(ret == false)
                {
                    DLOG("读取消息长度失败！");
                    return false;
                }
                offset += sizeof(size_t);
                // 再读取对应长度的数据
                std::string msg_body(msg_size, '\0');
                ret = _df_helper.Read(&msg_body[0], offset, msg_size);
                if(ret == false)
                {
                    DLOG("读取消息数据失败！");
                    return false;
                }
                offset += msg_size;
                MessagePtr msgp = std::make_shared<mmq::Message>();
                msgp->mutable_payload()->ParseFromString(msg_body); // 进行反序列化。注意：是payload进行反序列化
                // DLOG("加载到有效数据：%s", msgp->payload().body().c_str());
                // 数据无效则跳过
                if(msgp->payload().valid() == "0") continue;
                result.push_back(msgp);
            }   
            return true;
        }

        bool Insert(const std::string &filename, MessagePtr &msg)
        {   
            // 1. 获取msp序列化后的字符串
            std::string body = msg->payload().SerializeAsString();
            // 2. 获取文件长度
            FileHelper f_helper(filename);
            size_t offset = f_helper.Size();
            size_t msg_size = body.size();
            // 3. 将数据写入文件指定位置: 1. 先写入size_t字节数据长度  2. 再写入指定长度数据
            bool ret = f_helper.Write((char*)&msg_size, offset, sizeof(size_t));
            if(ret == false)
            {
                DLOG("向队列数据文件写入数据长度失败！");
                return false;
            }
            ret = f_helper.Write(body.c_str(), offset + sizeof(size_t), body.size());
            if(ret == false)
            {
                DLOG("向队列数据文件写入数据失败! ");
                return false;
            }
            // 3. 设置写入内容在文件中的偏移量和写入长度
            msg->set_offset(offset + sizeof(size_t));
            msg->set_length(body.size());
            return true;
        }

    private:
        std::string _qname;     // 队列名称
        std::string _datafile;  // 存储队列消息数据文件
        std::string _tmpfile;   // 临时文件(用于垃圾回收)
    };

    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;
    public:
        QueueMessage(std::string &basedir, const std::string &qname)
         : _mapper(basedir, qname), _qname(qname), _valid_count(0), _total_count(0)
        {}

        // 恢复历史消息
        bool Recovery()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs = _mapper.GC();
            for(auto &msg : _msgs)
            {
                _durable_msgs.insert({msg->payload().properties().id(), msg});
            }
            _valid_count = _total_count = _msgs.size();
            return true;
        }

        bool Insert(const BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            // 1. 构造MessagePtr对象并插入_msgs链表
            MessagePtr msg = std::make_shared<Message>();
            msg->mutable_payload()->set_body(body);
            // 根据bp和queue_is_durable共同判断是否持久化
            if(bp != nullptr)
            {
                DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UuidHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            std::unique_lock<std::mutex> lock(_mutex);

            // 2. 判断是否持久化
            if(msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                msg->mutable_payload()->set_valid("1"); // 设置消息为有效消息
                // 3. 将持久化消息写入文件(内部会自动设置偏移量和长度)
                bool ret = _mapper.Insert(msg); 
                if(ret == false)
                {
                    DLOG("持久化存储消息：%s 失败了！", body.c_str());
                    return false;
                }
                _valid_count++; 
                _total_count++;
                _durable_msgs.insert({msg->payload().properties().id(), msg});
            }
            _msgs.push_back(msg);
            return true;
        }

        // 获取并移除队头消息
        MessagePtr Front()
        {
            if(_msgs.size() == 0) return MessagePtr();
            std::unique_lock<std::mutex> lock(_mutex);
            // 1. 获取队头MessagePtr
            MessagePtr msg = _msgs.front();
            _msgs.pop_front();
            // 2. 将消息放入_waitack_msgs中
            _waitack_msgs.insert({msg->payload().properties().id(), msg});
            return msg;
        }
        
        // 消息被客户端确认后，将该消息从文件和内存中删除，每次删除后，判断是否需要垃圾回收
        bool Remove(const std::string &msg_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 1. 根据msg_id从_waitack_msgs中获取该消息
            auto it = _waitack_msgs.find(msg_id);
            if(it == _waitack_msgs.end())
            {
                DLOG("没有找到要删除的消息：%s", msg_id.c_str());
                return true;
            }
            MessagePtr msg = it->second;

            // 2. 判断是否持久化
            if(msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 3. 从文件中删除消息
                bool ret = _mapper.Remove(msg);
                if(ret == false)
                {
                    DLOG("从文件中删除消息 %s 失败！", msg->payload().properties().id().c_str());
                    return false;
                }
                _durable_msgs.erase(msg_id);
                _valid_count--;   // 持久化文件中有效消息数量减1
                // 4. 尝试垃圾回收
                GC();
            }
            // 5. 删除内存中的消息  
            _waitack_msgs.erase(msg_id);
            // DLOG("确认消息后，删除消息的管理成功: %s", msg->payload().body().c_str());
        }
        
        // 获取所有可推送消息数量
        size_t GetAbleCount()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }
        size_t TotalCount()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }
        size_t DurableCount()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }
        size_t WaitAckCount()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }
        void Clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.RemoveMsgFile();
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _valid_count = 0;
            _total_count = 0;
        }
    private:
        // 判断是否要进行垃圾回收(当总消息超过2000，并且有效消息不足一半时，进行垃圾回收)
        bool GCCheck()
        {
            if(_total_count > 2000 && _valid_count * 10 < _total_count * 5)
            {
                return true;
            }
            return false;
        }
        void GC()
        {
            // 1. 判断是否要进行垃圾回收
            if(GCCheck() == false) return;
            // 2. 进行垃圾回收并得到有效消息
            std::list<MessagePtr> msgs = _mapper.GC();
            for(auto &msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if(it == _durable_msgs.end())
                {
                    DLOG("垃圾回收后，有一条持久化消息，在内存中没有进行管理");
                    _msgs.push_back(msg);  // 做法：重新添加到推送链表的末尾
                    _durable_msgs.insert({msg->payload().properties().id(), msg});
                    continue;
                }
                // 3. 更新每一条消息的实际存储位置
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            // 4. 更新当前的有效消息和总的持久化消息数量
            _valid_count = _total_count = msgs.size();
        }
    private:
        std::mutex _mutex;
        std::string _qname;
        size_t _valid_count;          // 有效消息数量
        size_t _total_count;          // 总持久化消息数量(包括无效的持久化消息)
        MessageMapper _mapper;
        std::list<MessagePtr> _msgs;  // 待推送消息
        std::unordered_map<std::string, MessagePtr> _durable_msgs;  // 持久化消息hash
        std::unordered_map<std::string, MessagePtr> _waitack_msgs;  // 待确认消息hash
    };

    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;
    public:
        // 不在构造函数中直接获取队列信息，因为这需要包队列的头文件。设计函数在最后整合时传参来获取
        MessageManager(const std::string &basedir) : _basedir(basedir)
        {}
        void Clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for(auto &qmsg : _queue_msgs)
            {
                qmsg.second->Clear();
            }
            _queue_msgs.clear();
        }

        void InitQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it != _queue_msgs.end())
                {
                    return;
                }
                // 确认没有后才进行构造
                qmp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert({qname, qmp});
            }
            qmp->Recovery();
        }

        void DestroyQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    return;
                }
                qmp = it->second;
                _queue_msgs.erase(it);
            }
            qmp->Clear();
        }

        bool Insert(const std::string &qname, BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("向队列%s新增消息失败，找不到消息管理句柄！", qname.c_str());
                    return false;
                }
                qmp = it->second;
            }
            return qmp->Insert(bp, body, queue_is_durable);
        }

        MessagePtr Front(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s队首消息失败，找不到消息管理句柄！", qname.c_str());
                    return MessagePtr();
                }
                qmp = it->second;
            }
            return qmp->Front();
        }

        void Ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("确认队列%s消息%s，找不到消息管理句柄！", qname.c_str(), msg_id.c_str());
                    return;
                }
                qmp = it->second;
            }
            qmp->Remove(msg_id);
        }

        size_t GetAbleCount(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s待推送消息数量失败，找不到消息管理句柄！", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->GetAbleCount();
        }
        size_t TotalCount(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s总持久化消息数量失败，找不到消息管理句柄！", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->TotalCount();
        }
        size_t DurableCount(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s有效持久化消息数量失败，找不到消息管理句柄！", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->DurableCount();
        }
        size_t WaitAckCount(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s待确认消息数量失败，找不到消息管理句柄！", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->WaitAckCount();
        }

    private:
        std::mutex _mutex;
        std::string _basedir;   // 所有队列持久化消息存储文件的目录
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs;  // 对所有队列管理的句柄
    };

}
#endif
